package statistics;



import java.io.IOException;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ClickLogStatistic extends Configured implements Tool{
	public static final String DIMENSION_LIST = "DIMENSION_LIST";
	public static final String DIMENSION_SEP = "x";
	static enum Counters {PARSE_ERROR, TOO_LARGE_PAGE_NUM};
	
	public static class Map extends MapReduceBase implements Mapper<Text, Text, Text, LongWritable>{
		private int[] featrueList = null;
		
		private static final String KEY_TYPE = "c1";
		private static final LongWritable one = new LongWritable(1);
		private static final int COLUMN_COUNT = 27;
		
		@Override 
		public void configure(JobConf conf){
			String str = conf.get(DIMENSION_LIST);
			String[] dims = str.split(DIMENSION_SEP);
			featrueList = new int[dims.length];
			int i = 0;
			for (String dim : dims){
				featrueList[i++] = Integer.parseInt(dim) - 2;
			}
		}
		
		@Override
		public void map(Text key, Text value,
				OutputCollector<Text, LongWritable> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			if (!key.toString().equalsIgnoreCase(KEY_TYPE))
				return;
			
			StringBuilder builder = new StringBuilder();
			String[] columns = value.toString().split("\t");
			if (columns.length != COLUMN_COUNT - 1){
				reporter.incrCounter(Counters.PARSE_ERROR, 1);
				throw new IOException("parse error line: " + key.toString() + "\t" + value.toString());
			}
			int pageNum = Integer.parseInt(columns[15]);
			if (pageNum > 20){
				reporter.incrCounter(Counters.TOO_LARGE_PAGE_NUM, 1);
				return;
			}
				
			for (int i = 0; i < featrueList.length; ++i){
				builder.append(columns[featrueList[i]]);
				builder.append("\t");
			}
			output.collect(new Text(builder.toString()), one);			
		}
	}
	
	public static class Reduce extends MapReduceBase implements Reducer<Text, LongWritable, Text, LongWritable>{

		@Override
		public void reduce(Text key, Iterator<LongWritable> value,
				OutputCollector<Text, LongWritable> output, Reporter reporter)
				throws IOException {
			// TODO Auto-generated method stub
			long sum = 0;
			while (value.hasNext()){
				sum += value.next().get();
			}
			output.collect(key, new LongWritable(sum));
		}
		
	}
	
	@Override
	public int run(String[] args) throws Exception{
	
		JobConf conf = new JobConf(getConf(), ClickLogStatistic.class);
		conf.setJobName("ClickLog Statistic");
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(LongWritable.class);
		
		conf.setMapperClass(Map.class);
		conf.setReducerClass(Reduce.class);
		conf.setCombinerClass(Reduce.class);
		
		conf.setInputFormat(KeyValueTextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);
		
		
		String strDimensions = null;
		String input = null;
		String output = null;
			
		for (int i = 0; i < args.length; ++i){
			if (args[i].equalsIgnoreCase("-dim")){
				strDimensions = args[++i];
			}
			else if (args[i].equalsIgnoreCase("-i")){				
				input = args[++i];
			}
			else if (args[i].equalsIgnoreCase("-o")){
				output = args[++i];
			}
			else if (args[i].equalsIgnoreCase("-mc")){
				conf.setNumMapTasks(Integer.parseInt(args[++i]));
			}
			else if (args[i].equalsIgnoreCase("-rc")){
				conf.setNumReduceTasks(Integer.parseInt(args[++i]));
			} 
			else if (args[i].equalsIgnoreCase("-nm")){
				conf.setJobName("ClickLog Statistics" + args[++i]);
			}
		}
		if (strDimensions == null){
			System.out.println("dimension is null!");
			return 1;
		}
		if (input == null){
			System.out.println("input is null");
			return 1;
		}
		if (output == null){
			System.out.println("output is null");
			return 1;
		}
		
		conf.set(DIMENSION_LIST, strDimensions);
		FileInputFormat.setInputPaths(conf, new Path(input));
		FileOutputFormat.setOutputPath(conf, new Path(output));

		JobClient.runJob(conf);
		return 0;
	}

	public static void main(String[] args) throws Exception{
		System.out.println("start to run click log statistics2");
		System.out.println("input parameters: ");
		for (String s : args){
			System.out.println(s);
		}
		
		if (args.length < 6){
			System.out.println(args.length);
			System.out.println("usage: -dim <dim1xdim2> -i <intput> -o <output> [-mc <map count> -rc <reduce count> -nm <job name>]");
			System.exit(0);
		}
		
		int res = ToolRunner.run(new Configuration(), new ClickLogStatistic(), args);
		System.out.println("job commit status :" + res);
		System.exit(res);

	}

}


	

